package com.yanggu.bigdata.realtime.app.function

import cn.hutool.core.date.DateUtil
import cn.hutool.core.util.StrUtil
import com.alibaba.fastjson.JSONObject
import org.apache.flink.api.common.functions.RichFilterFunction
import org.apache.flink.api.common.state.{StateTtlConfig, ValueStateDescriptor}
import org.apache.flink.api.common.time.Time

import java.util.Date

class UniqueVisitRichFilterFunction extends RichFilterFunction[JSONObject] {

  private lazy val dateStrState = {
    val dateStrStateDescriptor = new ValueStateDescriptor[String]("dateStrState", classOf[String])
    val config = new StateTtlConfig.Builder(Time.days(1L))
      .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
      .build()
    dateStrStateDescriptor.enableTimeToLive(config)
    getRuntimeContext.getState(dateStrStateDescriptor)
  }

  /*
  {
    "common": {
      "ar": "110000",
      "uid": "38",
      "os": "Android 11.0",
      "ch": "xiaomi",
      "is_new": "1",
      "md": "Xiaomi 9",
      "mid": "mid_14",
      "vc": "v2.1.134",
      "ba": "Xiaomi"
    },
    "page": {
      "page_id": "payment",
      "item": "3,4,7",
      "during_time": 9777,
      "item_type": "sku_ids",
      "last_page_id": "trade"
    },
    "ts": 1608276037000
  }
 */
  override def filter(value: JSONObject): Boolean = {

    //取出pageId
    val pageId = value.getJSONObject("page").getString("page_id")
    //如果pageId不为空, 表示不是首次访问, 直接过滤掉
    if (StrUtil.isEmpty(pageId)) {
      return false
    }

    val currentDateStr = dateStrState.value()
    val dateStr = DateUtil.formatDate(new Date(value.getLong("ts")))

    //如果状态为空或者是日期不相等, 更新状态, 并且返回true
    if (StrUtil.isEmpty(currentDateStr) || !StrUtil.equals(currentDateStr, dateStr)) {
      dateStrState.update(dateStr)
      true
      //否则直接返回false
    } else {
      false
    }
  }

}
